package com.wpml.netty.rpc.client;

import com.wpml.netty.rpc.ResponseCallback;
import com.wpml.netty.rpc.common.MyBody;
import com.wpml.netty.rpc.common.MyHeader;
import com.wpml.netty.rpc.server.ServerDecode;
import com.wpml.netty.rpc.util.SerDerUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.io.*;
import java.net.*;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Description 连接工厂
 * @Date 2021-05-06
 * @Author wangpei
 **/
@SuppressWarnings("all")
public class ClientFactory {

    static String transType = "rpc";

    int poolSize = 5;

    private ClientFactory(){}
    private static final ClientFactory factory;

    private ConcurrentHashMap<InetSocketAddress, ClientPool> clientMap = new ConcurrentHashMap<>();

    private final Object lock = new Object();

    Random rand = new Random();

    static {
        factory = new ClientFactory();
    }

    public static ClientFactory getFactory() {
        return factory;
    }


    public static CompletableFuture<Object> transport(MyBody body) throws InterruptedException {

        CompletableFuture<Object> response = new CompletableFuture<>();
        transType = "http";

        if (transType.equals("rpc")) {
            byte[] msgBody = SerDerUtil.ser(body);
            // 封装头信息
            MyHeader header = MyHeader.createHeader(msgBody);
            byte[] msgHeader = SerDerUtil.ser(header);

            NioSocketChannel clientChannel = factory.getClient(new InetSocketAddress("localhost", 9090));
            ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.directBuffer(msgHeader.length + msgBody.length);


            // 注册回调
            ResponseCallback.addCallback(header.getRequestID(), response);

            byteBuf.writeBytes(msgHeader);
            byteBuf.writeBytes(msgBody);
            clientChannel.writeAndFlush(byteBuf);
        } else { // http协议
            urlTS(body, response);
        }

        return response;
    }

    private static void nettyTS(MyBody body, CompletableFuture<Object> response) {

    }

    private static void urlTS(MyBody body, CompletableFuture<Object> response) {
        System.out.println("==== urlTS ====");
        Object retObj = null;
        try {
            // 建立连接
            URL url = new URL("http://localhost:9090");
            HttpURLConnection httpConn = (HttpURLConnection) url.openConnection();
            httpConn.setRequestMethod("POST");
            httpConn.setDoOutput(true);
            httpConn.setDoInput(true);
            // 发送请求数据
            OutputStream out = httpConn.getOutputStream();
            ObjectOutputStream objOut = new ObjectOutputStream(out);
            objOut.writeObject(body);

            if (httpConn.getResponseCode() == HttpURLConnection.HTTP_OK) {
                // 读取服务端返回值
                InputStream in = httpConn.getInputStream();
                ObjectInputStream oin = new ObjectInputStream(in);
                MyBody ret = (MyBody) oin.readObject();
                retObj = ret.getRes();
            }
        } catch (IOException | ClassNotFoundException e) {
            e.printStackTrace();
        }
        response.complete(retObj);
    }


    public NioSocketChannel getClient(InetSocketAddress address) {
        ClientPool clientPool = clientMap.get(address);
        // 并发访问时控制重复覆盖
        if (clientPool == null) {
            synchronized (lock) {
                clientPool = clientMap.get(address);
                if (clientPool == null) {
                    clientMap.putIfAbsent(address,new ClientPool(poolSize));
                    clientPool =  clientMap.get(address);
                }
            }
        }

        // 随机获取一个连接
        int index = rand.nextInt(poolSize);
        NioSocketChannel client = clientPool.clients[index];
        if (client != null && client.isActive()) {
            return client;
        } else {
            // 创建连接
            synchronized (clientPool.lock[index]) {
                if (client == null || !client.isActive()) {
                    clientPool.clients[index] = createClient(address);
                }
            }
        }
        return clientPool.clients[index];
    }

    private NioSocketChannel createClient(InetSocketAddress address) {

        NioEventLoopGroup clientWorker = new NioEventLoopGroup(2);
        Bootstrap bs = new Bootstrap();
        ChannelFuture connect = bs.group(clientWorker)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline()
                                .addLast(new ServerDecode())
                                .addLast(new ClientResponseHandler());
                    }
                }).connect(address);

        try {
            return  (NioSocketChannel) connect.sync().channel();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

}
